/*
 * file_receiver.cpp
 *
 *  Created on: 2009-1-7
 *      Author: fanliangliang
 */

#include "file_receiver.h"
#include "file_sender.h"
#include "file_deleter.h"
#include "uncompressor.h"
#include "common_struct.h"
#include "common_packet.h"
#include <fstream>
#include <algorithm>
#include <memory> // for auto_ptr

using namespace std;
using namespace toxic;

FileReceiver::FileReceiver(FileSender* sender, Uncompressor* uncompressor)
:sender_(sender)
,uncompressor_(uncompressor) {

}

FileReceiver::~FileReceiver() {

}

void FileReceiver::process(TcpSocket* _socket) {
	if (_socket == 0) {
		xerror << "null pointer {_socket}, at {void FileReceiver::process(TcpSocket* socket)}";
		return;
	}

	auto_ptr<TcpSocket> socket(_socket);

	try {
		FileCountPacket fcp;
		if (!fcp.receiveVia(socket.get())) {
			xerror << "receive FileCountPacket failed";
			return;
		}

		uint16_t fileCount = fcp.fileCount();
		xdebug << "start receiving {" << fileCount << "} files";

		for (uint16_t i = 0; i < fileCount; ++i) {
			receiveFile(socket.get());
		}

		xdebug << "finish receiving {" << fileCount << "} files";

		JobOfferPacket jop;
		if (!jop.receiveVia(socket.get())) {
			xerror << "receive JobOfferPacket failed";
			return;
		}

		if (jop.moreJob()) {
			xtrace << "there are more receivers, notify sender";
			sender_->push(jop.jobArgument());
		}
		else {
			xtrace << "no more receivers, try delete temp tar.gz files";
			foreach (const string& filename, tempFiles_) {
				FileDeleter::instance()->releaseFile(filename, 1);
			}
		}
		tempFiles_.clear();
	}
	catch (const exception& e) {
		xerror << e.what();
	}
	catch (...) {
		xerror << "oops, caught an unexpected exception";
	}
}

bool FileReceiver::receiveFile(TcpSocket* socket) {
	FileArgumentPacket fap;
	if (!fap.receiveVia(socket)) {
		xerror << "receive FileArgumentPacket failed";
		return false;
	}

	auto_ptr<FileArgument> arg(fap.fileArgument());

	if (!toxic::createDirectory(arg->destination)) {
		xerror << "create directory {" << arg->destination << "} failed";
		return false;
	}

	string pathname = arg->destination + arg->name;
	ofstream ofs(pathname.c_str(), ios::binary);
	if (!ofs) {
		xerror << "create file {" << pathname << "} failed";
		return false;
	}

	xdebug << "start receiving file {" << pathname << "}, size {" << arg->size << "} bytes";

	char buffer[1024 * 1024] = { 0 };
	uint32_t nleft = arg->size;
	while (nleft > 0) {
		size_t length = std::min(nleft, sizeof(buffer));
		size_t nrecv = socket->receive(buffer, length);
		if (nrecv == 0) {
			xerror << "the connection has been gracefully closed, {" << nleft << "} bytes lost";
			break;
		}
		ofs.write(buffer, nrecv);
		nleft -= nrecv;
	}
	ofs.close();

	xdebug << "finish receiving file {" << pathname << "}, notify uncompressor";

	if (arg->isCompressed) {
		tempFiles_.push_back(pathname);
		FileDeleter::instance()->registerFile(pathname, 2);
	}

	// notify uncompressor
	uncompressor_->push(arg.release());

	xtrace << "start sending FileReceivedPacket";

	FileReceivedPacket frp;
	frp.sendVia(socket);

	xtrace << "finish sending FileReceivedPacket";

	return true;
}
